package com.yungui.kafka.consumer.service;

import com.yungui.kafka.consumer.entity.DiffData;
import com.yungui.kafka.consumer.entity.EsEntity;
import com.yungui.kafka.consumer.entity.FlatMessage;
import com.yungui.kafka.consumer.es.EsCommonService;
import com.yungui.kafka.consumer.util.JsonUtil;
import com.yungui.kafka.consumer.util.SnowFlake;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.elasticsearch.action.bulk.BulkResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.*;

@Component
public class KafkaConsumer {
    //日志
    Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private EsCommonService esCommonService;
    @Autowired
    private FlatMessageFilter flatMessageFilter;
    // 消费监听
    @KafkaListener(id = "${spring.kafka.consumer.client-id}",
            groupId = "${spring.kafka.consumer.group-id}",
            containerFactory = "batchFactory",
            topicPartitions = {@TopicPartition(topic = "${spring.kafka.template.default-topic}", partitions = {"0"})})
    public void onMessage(List<ConsumerRecord> recordList, Acknowledgment ack) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        logger.info("原生消息:{}" , recordList);
        List<FlatMessage> flatMessages = flatMessageFilter.filterMessage(recordList);
        if (!CollectionUtils.isEmpty(flatMessages)){
            SnowFlake snowFlake = new SnowFlake(0, 0);
            for (FlatMessage flatMessage : flatMessages) {
                Map<String,Object> jsonMap = new HashMap<>();
                jsonMap.put("id",String.valueOf(snowFlake.nextId()));
                jsonMap.put("table",flatMessage.getTable()==null?"":flatMessage.getTable());
                jsonMap.put("type",flatMessage.getType());
                jsonMap.put("data",flatMessage.getData());
                jsonMap.put("old",flatMessage.getOld());
                jsonMap.put("create_time",Calendar.getInstance().getTime());
                this.buildChangeDataList(flatMessage.getData(),flatMessage.getOld());
                EsEntity<Map> esEntity = new EsEntity<>();
                esEntity.setId(jsonMap.get("id").toString());
                esEntity.setData(jsonMap);
                logger.info("数据有变更如下：{}",JsonUtil.toJson(esEntity));
                BulkResponse bulkResponse = esCommonService.insert("data_change_es_"+flatMessage.getTable(),esEntity);
                logger.info("插入es结果:{},{}",bulkResponse.status().getStatus(),bulkResponse.buildFailureMessage());
            }
        }
        ack.acknowledge();
    }

    /**
     * 比较不同
     * @param data
     * @param old
     * @return
     */
    private List<List<DiffData>> buildChangeDataList(List<Map<String,Object>> data, List<Map<String,Object>> old) {
        if(data==null || old==null){
            return null;
        }
        List<List<DiffData>> changeDataList = new ArrayList<>();
        int index = 0;
        for (index=0 ; index < old.size() ; index++ ) {
            Map<String,Object> oldMap = old.get(index);
            Map<String,Object> newMap = data.get(index);
            List<DiffData> diffDataList = new ArrayList<>();
            for (String oldKey : oldMap.keySet()) {
                DiffData diffData = new DiffData(oldKey,oldMap.get(oldKey)==null?null:oldMap.get(oldKey).toString(),newMap.get(oldKey)==null?null:newMap.get(oldKey).toString());
                diffDataList.add(diffData);
            }
        }
        return changeDataList;
    }

}
